package com.codejiwei.core.rdd

import org.apache.spark.rdd.JdbcRDD
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{DriverManager, ResultSet}

object Spark_RDD_MySQL {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("RDDMySQL")
    val sc = new SparkContext(conf)

    // 配置MySQL连接信息
    val mysqlHost = "localhost"
    val mysqlPort = "3306"
    val mysqlDatabase = "mydb"
    val mysqlTable = "flink_score"
    val mysqlUser = "root"
    val mysqlPassword = "123456"

    // 定义MySQL连接函数
    def createConnection(): java.sql.Connection = {
      Class.forName("com.mysql.cj.jdbc.Driver")
      DriverManager.getConnection(s"jdbc:mysql://$mysqlHost:$mysqlPort/$mysqlDatabase", mysqlUser, mysqlPassword)
    }

    // 定义查询语句
    val sqlQuery = "SELECT * FROM " + mysqlTable + " WHERE id >= ? AND id <= ?"
    //TODO 必须设置上边界和下边界 两个占位符 下面sql会报错
    val sqlQuery1 = "SELECT * FROM " + mysqlTable

    // 定义RDD
    val jdbcRDD = new JdbcRDD(
      sc,
      createConnection,
      sqlQuery,
      lowerBound = 1,
      upperBound = 100,
      numPartitions = 2,
      mapRow = (resultSet: ResultSet) => resultSet.getString("name") // 替换 "column_name" 为实际的列名
    )

    // 打印RDD的内容
    jdbcRDD.collect().foreach(println)

    // 关闭SparkContext
    sc.stop()

  }
}
